#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include "MQTT/MQTT.h"
#include "MQTT/Params.h"
#include "MQTT/MQTTError.h"

static int8_t enableRecvData = 0;
static int8_t enableProcess = 0;
//static const char* clientPemContent = "";	//安全通道证书请联系商务获取
static const char* clientPemContent = "-----BEGIN CERTIFICATE-----\n"
                                      "MIIDiTCCAnGgAwIBAgIEVAjk1TANBgkqhkiG9w0BAQsFADB1MRgwFgYDVQQDEw93\n"
                                      "d3cudW5pLXViaS5jb20xDzANBgNVBAsTBnVuaXViaTEVMBMGA1UECgwMVW5pVWJp\n"
                                      "77yMSW5jMREwDwYDVQQHEwhIYW5nWmhvdTERMA8GA1UECBMIWmhlSmlhbmcxCzAJ\n"
                                      "BgNVBAYTAkNOMB4XDTE5MDkyMDExNTcxNVoXDTQ5MDkxMjExNTcxNVowdTEYMBYG\n"
                                      "A1UEAxMPd3d3LnVuaS11YmkuY29tMQ8wDQYDVQQLEwZ1bml1YmkxFTATBgNVBAoM\n"
                                      "DFVuaVViae+8jEluYzERMA8GA1UEBxMISGFuZ1pob3UxETAPBgNVBAgTCFpoZUpp\n"
                                      "YW5nMQswCQYDVQQGEwJDTjCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEB\n"
                                      "ALM6WDomjbTre1fj/6LLFXnggdDWxKv6pawIXQcFAxKcZi+84NijxKx9hQoyDxpm\n"
                                      "AhB35xfrJVKw08A6uRvTf/p4yMT7mctwb/RkntYVfOLMlqKNmTGjT78gySGMLY6P\n"
                                      "7IETf3Y7DzkIE+MgfuFxI93mI+l4P2+UCbkYH5oldHvST30olYxpOPzauLQecG4M\n"
                                      "lMXi8MKpiDcq6rWGEo64HeiXR2qGLm6KZ7qtTlyy1a0hy1kNT8vdrtVDLnTHuFPh\n"
                                      "6NR4VuuJwwFvLjKyNluDys+hyAWPgtRL5UPEpCLJMRbxmjcq7LmQW5ynx6tlyYUj\n"
                                      "ucBV3rSU6yWJ0ORjegDaazMCAwEAAaMhMB8wHQYDVR0OBBYEFG0E+N1sI6lDdNrw\n"
                                      "Y9NTaafCjiVXMA0GCSqGSIb3DQEBCwUAA4IBAQBMGKJ0lswoYA2WpxtQmHklCGav\n"
                                      "SrjeNKVSKGVOicMx7s2ZS6Eb3A9geasfh3GoK/cIs+KEeOV1AXi5mdUnlNWWSmnq\n"
                                      "RmFx/a6pPY473FvSf6Jnu56VkKnkC2TeFYnAM3Olw8sm6vvVJPgBNhbD6USjglpb\n"
                                      "8ULb5fHsZrj/kZoedwgbGPKflroioQG6o93/2xch08pyQbPTT20WcvWKBvu+F5jB\n"
                                      "hh8Vd56ZQ9IyVCoI+hRzW2JnrCm5NYwNUBVAbdYQEedZvnH9mJXenm5yx0YPWHcZ\n"
                                      "PbKB2jmaGPM9FNKwzomGkkqLXhGGq6MfNEjjNXdejVlqmhFzFyPbfhXM+mUr\n"
                                      "-----END CERTIFICATE-----\n";

static void* mqttProcessThread(void *args) {

    int32_t res = aiotSuccess;
    while (enableProcess) {

        res = UMQTT_ProcessReTransport(args);
        if (res == disableExecute) {
            break;
        }

        sleep(1);
    }

    return nullptr;
}
static void* mqttRecvThread(void* args) {

    int32_t res = aiotSuccess;
    while (enableRecvData) {
        res = UMQTT_HandleData(args);
        if (res == disableExecute) {
            break;
        }

        sleep(1);
    }

    return nullptr;
}

/**日志回调，不需要改动*/
static void loggerCb(AiotLoggerLevel level,char* message) {
    printf("[aiot] message %s",message);
}

/**接收MQTT 消息*/
static void recvMQTTMessage(void *handle, const RecvMQTTMessage* packet, void *context) {

    switch (packet->type) {
        case recvHeartBeatRes: {
            printf("ping response \n");
        }
            break;

        case recvSubscribeAck: {
            printf("subscribe ack, res: %d, packet id: %d, max qos: %d\n",packet->data.subscribeAck.res, packet->data.subscribeAck.packetId, packet->data.subscribeAck.maxQos);
        }
            break;
        case recvUnSubscribeAck: {
            printf("unsubscribe ack, packet id: %d\n",packet->data.unsubscribeAck.packetId);
        }
            break;
        case recvPublish: {
            int32_t idx = 0;
            printf("publish, qos: %d, topic: %d.%s\n", packet->data.publish.qos, packet->data.publish.topicLen, packet->data.publish.topic);
            printf("publish, payload: %d.%s\n", packet->data.publish.payloadLen, packet->data.publish.payload);
            printf("publish, payload len: %d\n", packet->data.publish.payloadLen);

            if (nullptr != packet->data.publish.pubProp) {

                printf("publish, topic alias is %d\n", packet->data.publish.pubProp->topicAlias);

                if (packet->data.publish.pubProp->subIdentifier > 0) {
                    printf("publish, subscription identifer: %d\n", packet->data.publish.pubProp->subIdentifier);
                }

                if (packet->data.publish.pubProp->responseTopic.len > 0) {
                    printf("response topic is %d.%s\n", packet->data.publish.pubProp->responseTopic.len,
                           packet->data.publish.pubProp->responseTopic.value);
                }
                if (packet->data.publish.pubProp->correlationData.len > 0) {
                    printf("correlation data is %d.%s\n", packet->data.publish.pubProp->correlationData.len,
                           packet->data.publish.pubProp->correlationData.value);
                }

                for (idx = 0; idx < MAX_USER_PROPERTY; idx++) {
                    UserProperty *user_prop = packet->data.publish.pubProp->userProperty[idx];
                    if (nullptr != user_prop) {
                        printf("publish, user property key: %d.%s\n", user_prop->key.len, user_prop->key.value);
                        printf("publish, user property value: %d.%s\n", user_prop->value.len, user_prop->value.value);
                    }
                }
            }
            printf("\n");
        }
            break;

        case recvPublishAck: {
            printf("publish ack, packet id: %d\n", packet->data.publishAck.packetId);
        }
            break;
        case recvConnectAck: {
            int idx = 0;
            printf("\n\n");
            printf("connect ack, topic alias max in server side: %d\n", packet->data.connectAck.property.maxTopicAlis);
            printf("connect ack, max qos: %d\n", packet->data.connectAck.property.maxQos);
            printf("connect ack, server assigned cid: %s\n", (char *)(packet->data.connectAck.property.clientId));
            printf("connect ack, reason code: %d\n", packet->data.connectAck.reasonCode);
            printf("connect ack, max_packet_size: %d\n", packet->data.connectAck.property.maxPacket);
            printf("connect ack, shared_subscription_available: %d\n", packet->data.connectAck.property.sharedSubAvailable);
            printf("connect ack, wildcard_subscription_available: %d\n", packet->data.connectAck.property.wildcardSubAvailable);
            printf("connect ack, subscription_identifier_available: %d\n",packet->data.connectAck.property.subIdentifierAvailable);
            printf("connect ack, server_receive_max: %d\n", packet->data.connectAck.property.serverReceiveMax);

            for (idx = 0; idx < MAX_USER_PROPERTY; idx++) {
                UserProperty *user_prop = packet->data.connectAck.property.userProperty[idx];
                if (nullptr != user_prop) {
                    printf("connect ack, user property key: %d.%s\n", user_prop->key.len, user_prop->key.value);
                    printf("connect ack, user property value: %d.%s\n", user_prop->value.len, user_prop->value.value);
                }
            }
        }
            printf("\n");
            break;
        case recvDisconnect: {
            printf("server disconnect, reason code: 0x%x\n", packet->data.disconnect.reasonCode);
        }
            break;
        default: {

        }
    }
}

/**AIOT 事件处理*/
static void aiotEventHandler(void *handle, const AiotEvent *event, void *context) {

    switch (event->type) {

        case aiotConnect: {
            printf("aiot mqtt connected\n");
        }
            break;
        case aiotReConnect: {
            printf("aiot mqtt reconnect\n");
        }
            break;

        case aiotDisconnect: {
            char *cause = (event->data.disconnect == networkDisconnect) ? ("network exception") :("heartbeat timeout");
            printf("aiot disconnect,reason : %s\n", cause);
        }
            break;

        default:
            break;
    }
}

/**********************业务测试**********************/

/**测试事件发布*/
static int32_t testPublish(void *entity, AiotDeviceInfo *access) {

    int32_t res = aiotSuccess;

    char topic[128] = {0};
    /**定义topic*/
	snprintf(topic, sizeof(topic) - 1,"xxxx");   /**自行定义发布内容*/

    /**如有需要组装业务消息体*/
    //char *payload = "{\"requestId\":\"110\",\"msgId\":1623926939410,\"timestamp\":1623926939410,\"version\":\"1.0\",\"identifier\":\"config\",\"method\":\"thing.config.get\",\"params\":{\"config\":1}}";

	PublishParams params = {topic,"",0,0,nullptr};
	
    /**发布消息**/
    res = UMQTT_Publish(entity,&params);
    if (res != aiotSuccess) {
        printf("publish failed, res:%d\n", res);
        return -1;
    }


    printf("test publish event done\n");
    return 0;
}

/**测试消息订阅**/
static int32_t testSubcrible(void* entity,AiotDeviceInfo* access) {

    int32_t res = aiotSuccess;

    /**定义topic*/
    char topic[128] = {0};
	
	snprintf(topic, sizeof(topic) - 1,"xxxx");  /**自行定义订阅内容*/
    /**构建订阅消息,并发送出去*/
    SubscribeParams params = {topic,1,nullptr,nullptr};

    /**参数2可以按照topic 定制回调*/
    res = UMQTT_Subscribe(entity,nullptr,&params);
    if (res != aiotSuccess) {
        printf("subscribe failed, res:%d\n",res);
        return -1;
    }

    return 0;
}

/**测试取消订阅*/
static int32_t testUnSubscribe(void* entity,AiotDeviceInfo* access) {

    int32_t res = aiotSuccess;
    char topic[128] = {0};
    snprintf(topic, sizeof(topic) - 1,"xxxx");  /**自行定义反订阅内容*/

    UnSubscribeParams params = {topic,nullptr};
    res = UMQTT_UnSubscribe(entity,&params);
    if (res != aiotSuccess) {
        printf("unsubscribe failed, res:%d\n",res);
        return -1;
    }

    return 0;
}

/**测试sdk 断开连接**/
static int32_t testDisconnect(void* entity) {

    int32_t res = aiotSuccess;
    printf("begin mqtt disconnect test\n");

    DisconnectParams params = {1,nullptr};

    /**发送断开连接参数*/
    res = UMQTT_Disconnect(entity,&params);
    if (res != aiotSuccess) {
        printf("disconnect failed, res:%d\n",res);
        return -1;
    }

    printf("end disconnect mqtt test\n");
    return 0;
}



/**sdk 使用主体逻辑**/
int32_t main(int32_t argc,char* argv[]) {

    int32_t ret = 0;
    pthread_t mqttProcessThreadId;
    pthread_t mqttRecvThreadId;

    /******************以下是sdk 初始化、参数配置逻辑************************/

    /**设置服务信息;非安全通道端口 1883,安全通道端口8883*/
    AiotServerAddress serverAddress = {"114.67.107.81",50990};

    /****************一机一密接入平台的配置参数******************************/
    /**1. productKey;2.deviceName,3.secretKey*/
    AiotDeviceInfo accessInfo = {"d0UC89mJDtDYC3tX","84E0F42AA9D91801","c9d022f4662142a283a91654d9efb7b2"};

    /*********************************通道配置***********************************/
    AiotChannelParams channelParams = {30*1000,3,1,3*1000,10*1000,10*1000,10*1000};

    /*****************************安全通道证书参数*********************************/
    AiotCredInfo aiotCredInfo;
    memset(&aiotCredInfo,0x00, sizeof(AiotCredInfo));
    aiotCredInfo.option = aiotCredFile;
    aiotCredInfo.tlsFragment = 16384;

    /**设置证书*/
	aiotCredInfo.credFile = clientPemContent;
    aiotCredInfo.credLength = strlen(clientPemContent);

    /*****************************************************************************/
    UMQTT_InitialGlobalEnv();
    UMQTT_SetLoggerCb(loggerCb);

    void* mqttEntity = UMQTT_CreateEntity(nullptr);

    UMQTT_SetServerInfo(mqttEntity,&serverAddress);
    UMQTT_SetDeviceInfo(mqttEntity,&accessInfo);
    UMQTT_SetChannelParams(mqttEntity,&channelParams);
    UMQTT_SetEventHandle(mqttEntity, aiotEventHandler);

    /**测试安全通道,非安全通道不要设置*/
    UMQTT_SetChannelCred(mqttEntity,&aiotCredInfo);
    UMQTT_SetRecvHandle(mqttEntity, recvMQTTMessage);
    UMQTT_SetDestroyTimeout(mqttEntity,12*1000);

    /**********************开始连接***************************/
    ret = UMQTT_ConnetServer(mqttEntity, nullptr);
    if (ret != aiotSuccess) {
        UMQTT_DestroyEntity(mqttEntity);
        printf("connect mqtt server failed: %d\n", ret);
        return -1;
    }

	enableRecvData = 1;
    ret = pthread_create(&mqttProcessThreadId, nullptr, mqttProcessThread,mqttEntity);
    if (ret < 0) {
        printf("pthread_create mqtt process thread failed: %d\n", ret);
        return -1;
    }

    enableProcess = 1;
    ret = pthread_create(&mqttRecvThreadId, nullptr, mqttRecvThread,mqttEntity);
    if (ret < 0) {
        printf("pthread_create mqtt recv thread failed: %d\n", ret);
        return -1;
    }
	
    /***TODO********************业务测试部分，可以在此添加业务测试逻辑*****************************/
    /**订阅事件*/
    testSubcrible(mqttEntity,&accessInfo);
	sleep(2);

    /**发布事件*/
    testPublish(mqttEntity,&accessInfo);
    sleep(5);

    /**取消订阅*/
    testUnSubscribe(mqttEntity,&accessInfo);
    sleep(5);

    testDisconnect(mqttEntity);

    int32_t count = 0;

    /**测试sdk退出逻辑,30s 后测试sdk销毁接口*/
    while (count++ < 30) {
        sleep(1);
    }

    enableRecvData = 0;
    enableProcess  = 0;
    /**销毁mqtt sdk 实体*/
    ret = UMQTT_DestroyEntity(&mqttEntity);
    if (ret != aiotSuccess) {
        printf("UMQTT_DestroyEntity failed: %04x\n", ret);
        return -1;
    }

    pthread_join(mqttProcessThreadId, nullptr);
    pthread_join(mqttRecvThreadId, nullptr);

    /**去初始化sdk全局环境信息**/
    UMQTT_DeinitialGlobalEnv();
    return 0;
}


